package com.prac.mq.original;

import com.prac.mq.model.MessageModel;

import java.util.Arrays;
import java.util.Date;
import java.util.List;

import java.util.function.Consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.junit.Test;

/**
 * mq基本用法
 *
 * @author H
 */
@Slf4j
public class MqTest {

    /**
     * 普通消息
     */
    @Test
    public void sampleProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("simple-group", false);
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR2);
        //producer.setMqClientApiTimeout(1000*10);
        producer.start();
        for (int i = 0; i < 10; i++) {
            Message message = new Message("simple-topic", "学习mq来了".getBytes());
            SendResult send = producer.send(message, 1000 * 10);
            log.info("消息状态：{},消息体：{}", send.getSendStatus(), new String(message.getBody()));
        }
        producer.shutdown();

    }

    @Test
    public void sampleConsumer() throws Exception {
        /**
         * 消费者组内的的消费者订阅关系必须保持一致
         */
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("simple-group", false);
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR2);
        //订阅主题 * 标识订阅主题中的所有消息 （可以弄消息过滤）
        consumer.subscribe("simple-topic", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                log.info("消费消息内容：" + new String(list.get(0).getBody()));
                log.info("消费上下文" + context);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.in.read();

    }

    /**
     * 异步消息
     */
    @Test
    public void asyncMq() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("async-group");
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR2);
        producer.start();
        Message message = new Message("async-topic", "异步消息".getBytes());
        producer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("发送成功" + sendResult.getSendStatus());
            }

            @Override
            public void onException(Throwable e) {
                log.info("发送失败" + e.getMessage());
            }
        });
        log.info("主线程执行");
        System.in.read();

    }

    /**
     * 单向消息：只负责发送，不关心结果；吞吐量大，有消息丢失风险；
     * 如日志消息的发送
     *
     * @throws Exception
     */
    @Test
    public void oneWayMq() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("test-oneway");
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR2);
        producer.start();
        Message message = new Message("oneway-topic", "单向消息".getBytes());
        producer.sendOneway(message);
        producer.shutdown();
    }


    /**
     * 延迟消息
     * 如订单确认后未付款需要在15min内确认
     *
     * @throws Exception
     */
    @Test
    public void delayMq() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("test-delay");
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR2);
        producer.start();
        Message message = new Message("delay-topic", "延迟消息".getBytes());
        //延迟级别
        message.setDelayTimeLevel(3);
        producer.send(message);
        log.info("当前时间：{}", new Date());
        producer.shutdown();
    }

    /**
     * 批量发送消息
     *
     * @throws Exception
     */
    @Test
    public void listMq() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("batch-group");
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR2);
        producer.start();
        List<Message> messageList = Arrays.asList(
                new Message("batch-topic", "批量消息".getBytes()),
                new Message("batch-topic", "批量消息2".getBytes()),
                new Message("batch-topic", "批量消息3".getBytes()),
                new Message("batch-topic", "批量消息4".getBytes()),
                new Message("batch-topic", "批量消息5".getBytes())
        );
        producer.send(messageList);
        producer.shutdown();
    }

    /**
     * 顺序消息
     * 如下单-付款-物流
     *
     * @throws Exception
     */
    @Test
    public void orderlyMq() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("orderly-group", false);
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR2);
        //producer.setMqClientApiTimeout(1000*10);
        producer.start();
        List<MessageModel> messageList = Arrays.asList(
                new MessageModel("1001", "user1", "u1-顺序消息2"),
                new MessageModel("1001", "user1", "u1-顺序消息1"),

                new MessageModel("1002", "user2", "u2-顺序消息1"),
                new MessageModel("1002", "user2", "u2-顺序消息2"),
                new MessageModel("1002", "user2", "u2-顺序消息3"),

                new MessageModel("1003", "user3", "u3-顺序消息1"),
                new MessageModel("1003", "user3", "u3-顺序消息2"),
                new MessageModel("1003", "user3", "u3-顺序消息3"),

                new MessageModel("1004", "user4", "u4-顺序消息1"),
                new MessageModel("1004", "user4", "u4-顺序消息2"),
                new MessageModel("1004", "user4", "u4-顺序消息3"),

                new MessageModel("1005", "user5", "u5-顺序消息1"),
                new MessageModel("1005", "user5", "u5-顺序消息2"),
                new MessageModel("1005", "user5", "u5-顺序消息3")
        );
        for (MessageModel t : messageList) {
            Message message = new Message("orderly-topic", t.toString().getBytes());
            try {
                producer.send(message, new MessageQueueSelector() {
                    /**
                     * 选择队列
                     * - 控制发送的顺序消息只依次发送到同一个queue中，消费的时候只从这个queue上依次拉取，则就保证了顺序
                     * - 当发送和消费参与的queue只有一个，则是全局有序；如果多个queue参与，则为分区有序，即相对每个queue，消息都是有序的
                     */
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        log.info("当前线程" + Thread.currentThread().getId() + "===>" + new String(msg.getBody()));
                        //当前主题有多少队列
                        int size = mqs.size();
                        /**
                         * arg是外层方法的参数3；即 t.getOrderId()
                         */
                        return mqs.get(arg.toString().hashCode() % size);
                    }
                }, t.getOrderId(), 1000 * 10);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
        log.info("发送成功");
        producer.shutdown();
    }

    @Test
    public void orderlyMqConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderly-consumer");
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR2);
        consumer.subscribe("orderly-topic", "*");
        consumer.setConsumeTimeout(1000 * 10);
        consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
            log.info("当前线程" + Thread.currentThread().getId() + "===>" + new String(msgs.get(0).getBody()));
            return ConsumeOrderlyStatus.SUCCESS;
        });
        consumer.start();
        System.in.read();
    }

    /**
     * 事务消息
     */
    @Test
    public void transMq() throws Exception {
        TransactionMQProducer producer = new TransactionMQProducer("trans-group");
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR2);
        producer.setTransactionListener(new TransactionListener() {
            /**
             * 执行本地业务方法
             * - 返回 UNKNOW 表示需要MQ回查才能确定状态；那么过一会代码会走下面的checkLocalTransaction(msg)方法
             * - arg是生产者发送事务消息时的本地事务逻辑
             *
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                Consumer<String> consumer;
                if (arg instanceof Consumer) {
                    consumer = (Consumer<String>) arg;
                    /**
                     * 本地事务执行成功，就会提交半事务消息
                     */
                    consumer.accept(new String(msg.getBody()));
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                return LocalTransactionState.UNKNOW;
            }

            /**
             * 回查本地事务状态；不执行业务操作，而是去确认上面业务操作是否有结果
             * - 默认是1min回查默认回查15次超过次数则丢弃打印日志可以通过参数设置
             * - transactionTimeOut 超时时间
             * - transactionCheckMax 最大回查次数
             * - transactionCheckInterval 回查间隔时间单位毫秒
             * ---
             * 触发条件
             * 1.当上面执行本地事务返回结果 UNKNOW 时,或者下面的回查方法也返回 UNKNOW 时会触发回查
             * 2.当上面操作超过20s没有做出一个结果，也就是超时或者卡住了，也会进行回查
             *
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                //本地事务执行结果状态放入redis中，mq事务回查时直接从redis中获取结果以此来确认事务状态
                log.info("本地事务回查：msg body:{}", new String(msg.getBody()));
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        producer.start();
        Message msg = new Message("trans-topic", "事务消息trans-msg".getBytes());
        /**
         * 方法参数arg是用来执行本地事务逻辑的
         */
        producer.sendMessageInTransaction(msg, new Consumer<String>() {
            @Override
            public void accept(String s) {
                log.info("执行本地事务逻辑：{}", s);
            }
        });
        System.in.read();

    }

    @Test
    public void transMqConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("trans-consumer");
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR2);
        consumer.subscribe("trans-topic", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                log.info("当前线程" + Thread.currentThread().getId() + "===>" + new String(msgs.get(0).getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.in.read();

    }

}
